#include "MQTTClient.h"
#include <iomanip>
#include <sstream>
#include <stdexcept>

namespace mqtt {

/**
 * @brief 构造函数
 * @param baseClientId 客户端基础ID，用于生成唯一客户端ID
 * @param cleanSession 是否使用干净会话（true：不保留上次会话信息）
 *
 * 初始化 mosquitto 库并生成客户端ID
 */
MQTTClient::MQTTClient(const std::string &baseClientId, bool cleanSession)
    : m_mosq(nullptr), m_connected(false), m_shouldReconnect(false),
      m_cleanSession(cleanSession), m_baseClientId(baseClientId) {
  mosquitto_lib_init();
  m_clientId = generateClientId();
}

/**
 * @brief 析构函数
 *
 * 断开连接并清理 mosquitto 库资源
 */
MQTTClient::~MQTTClient() {
  disconnect();
  mosquitto_lib_cleanup();
}

/**
 * @brief 连接到 MQTT Broker
 * @param host broker IP 或域名
 * @param port broker 端口
 * @param username 用户名（可选）
 * @param password 密码（可选）
 * @param qos 默认 QoS 值
 * @param keepalive 心跳间隔（秒）
 *
 * 异步连接 broker，并允许自动重连
 */
void MQTTClient::connect(const std::string &host, uint16_t port,
                         const std::string &username,
                         const std::string &password, uint8_t qos,
                         int keepalive) {
  std::lock_guard<std::mutex> lock(m_mutex);

  m_host = host;
  m_port = port;
  m_username = username;
  m_password = password;
  m_qos = qos;
  m_keepalive = keepalive;

  internalConnect();
  m_shouldReconnect = true;
}

/**
 * @brief 断开与 broker 的连接
 *
 * 停止消息循环，销毁 mosquitto 实例
 */
void MQTTClient::disconnect() {
  std::lock_guard<std::mutex> lock(m_mutex);
  m_shouldReconnect = false;

  if (m_mosq) {
    mosquitto_disconnect(m_mosq);
    mosquitto_loop_stop(m_mosq, false);
    mosquitto_destroy(m_mosq);
    m_mosq = nullptr;
  }

  m_connected = false;
}

/**
 * @brief 发布消息到指定主题
 * @param topic 主题
 * @param payload 消息内容
 * @param qos 消息 QoS，-1 表示使用默认值
 * @return MOSQ_ERR_SUCCESS 或错误码
 */
int MQTTClient::publish(const std::string &topic, const std::string &payload,
                        int qos) {

  int retry = 0;
  while (!m_connected && retry < 50) { // 最多等待 5 秒
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    retry++;
  }
  std::lock_guard<std::mutex> lock(m_mutex);

  if (!m_connected || !m_mosq) {
    std::cerr << "⚠️  未连接到 MQTT Broker\n";
    return MOSQ_ERR_NO_CONN;
  }

  int useQos = (qos >= 0) ? qos : m_qos;
  int ret = mosquitto_publish(m_mosq, nullptr, topic.c_str(), payload.size(),
                              payload.c_str(), useQos, false);
  if (ret != MOSQ_ERR_SUCCESS) {
    handleMosquittoError(ret);
  }

  return ret;
}

/**
 * @brief 订阅主题
 * @param topic 要订阅的主题
 * @param qos QoS，-1 表示使用默认值
 * @param cb 消息回调函数（可选）
 * @return true 成功，false 失败
 */
bool MQTTClient::subscribe(const std::string &topic, int qos,
                           MessageCallback cb) {
  std::lock_guard<std::mutex> lock(m_mutex);

  if (!m_connected || !m_mosq)
    return false;

  int useQos = (qos >= 0) ? qos : m_qos;
  int ret = mosquitto_subscribe(m_mosq, nullptr, topic.c_str(), useQos);
  if (ret != MOSQ_ERR_SUCCESS) {
    handleMosquittoError(ret);
    return false;
  }

  if (cb)
    m_topic_callbacks[topic] = cb;

  return true;
}

/**
 * @brief 为指定主题设置回调函数
 * @param topic 主题
 * @param cb 回调函数
 */
void MQTTClient::set_message_callback(const std::string &topic,
                                      MessageCallback cb) {
  std::lock_guard<std::mutex> lock(m_mutex);
  m_topic_callbacks[topic] = cb;
}

/**
 * @brief 设置默认回调函数
 * @param cb 回调函数
 */
void MQTTClient::set_default_callback(MessageCallback cb) {
  std::lock_guard<std::mutex> lock(m_mutex);
  m_default_cb = cb;
}

/**
 * @brief 检查客户端是否已连接
 * @return true 已连接，false 未连接
 */
bool MQTTClient::isConnected() const { return m_connected.load(); }

/**
 * @brief 生成客户端唯一ID
 * @return 客户端ID字符串
 */
std::string MQTTClient::generateClientId() const {
  if (m_baseClientId.empty()) {
    return getFormattedCurrentDateTime();
  }
  return m_baseClientId + "_" + getFormattedCurrentDateTime();
}

/**
 * @brief 获取格式化当前时间（年月日时分秒毫秒）
 * @return 格式化字符串
 */
std::string MQTTClient::getFormattedCurrentDateTime() {
  auto now = std::chrono::system_clock::now();
  auto now_time = std::chrono::system_clock::to_time_t(now);
  auto now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
      now.time_since_epoch());
  auto tm = *std::localtime(&now_time);

  char buffer[64];
  std::strftime(buffer, sizeof(buffer), "%Y%m%d%H%M%S", &tm);
  std::stringstream ss;
  ss << buffer << std::setfill('0') << std::setw(3) << now_ms.count() % 1000;
  return ss.str();
}

/**
 * @brief 内部连接 broker
 *
 * 创建 mosquitto 实例，设置回调，启动异步连接和消息循环
 */
void MQTTClient::internalConnect() {
  if (m_mosq) {
    mosquitto_destroy(m_mosq);
  }

  m_clientId = generateClientId();
  m_mosq = mosquitto_new(m_clientId.empty() ? nullptr : m_clientId.c_str(),
                         m_cleanSession, this);
  if (!m_mosq) {
    throw std::runtime_error("❌ 创建 mosquitto 实例失败");
  }

  mosquitto_connect_callback_set(m_mosq, onConnect);
  mosquitto_disconnect_callback_set(m_mosq, onDisconnect);
  mosquitto_message_callback_set(m_mosq, onMessageStatic);

  if (!m_username.empty() &&
      mosquitto_username_pw_set(m_mosq, m_username.c_str(),
                                m_password.c_str()) != MOSQ_ERR_SUCCESS) {
    mosquitto_destroy(m_mosq);
    m_mosq = nullptr;
    throw std::runtime_error("❌ 设置用户名/密码失败");
  }

  int ret =
      mosquitto_connect_async(m_mosq, m_host.c_str(), m_port, m_keepalive);
  if (ret != MOSQ_ERR_SUCCESS) {
    mosquitto_destroy(m_mosq);
    m_mosq = nullptr;
    handleMosquittoError(ret);
    return;
  }

  ret = mosquitto_loop_start(m_mosq);
  if (ret != MOSQ_ERR_SUCCESS) {
    mosquitto_destroy(m_mosq);
    m_mosq = nullptr;
    handleMosquittoError(ret);
  }
}

/**
 * @brief 处理 mosquitto 错误码
 * @param code mosquitto 错误码
 */
void MQTTClient::handleMosquittoError(int code) {
  switch (code) {
  case MOSQ_ERR_SUCCESS:
    break;
  case MOSQ_ERR_NO_CONN:
  case MOSQ_ERR_CONN_LOST:
    m_connected = false;
    startReconnect();
    break;
  default:
    std::cerr << "⚠️ MQTT 错误码: " << code << std::endl;
    break;
  }
}

/**
 * @brief 启动重连线程（异步）
 */
void MQTTClient::startReconnect() {
  if (!m_shouldReconnect)
    return;

  std::thread([this]() {
    std::this_thread::sleep_for(std::chrono::seconds(5));
    std::lock_guard<std::mutex> lock(m_mutex);
    if (m_shouldReconnect && !m_connected) {
      try {
        internalConnect();
      } catch (...) {
      }
    }
  }).detach();
}

/**
 * @brief 连接回调
 */
void MQTTClient::onConnect(struct mosquitto *, void *obj, int rc) {
  MQTTClient *client = static_cast<MQTTClient *>(obj);
  if (rc == MOSQ_ERR_SUCCESS) {
    client->m_connected = true;
    std::cout << "✅ 连接成功\n";
  } else {
    client->m_connected = false;
    client->handleMosquittoError(rc);
  }
}

/**
 * @brief 断开回调
 */
void MQTTClient::onDisconnect(struct mosquitto *, void *obj, int rc) {
  MQTTClient *client = static_cast<MQTTClient *>(obj);
  client->m_connected = false;
  client->handleMosquittoError(rc);
  std::cout << "⚠️ 已断开连接\n";
}

/**
 * @brief 静态消息回调，用于 mosquitto API
 */
void MQTTClient::onMessageStatic(struct mosquitto *, void *obj,
                                 const struct mosquitto_message *msg) {
  MQTTClient *client = static_cast<MQTTClient *>(obj);
  client->onMessage(msg);
}

/**
 * @brief 消息处理
 * @param msg 消息结构体
 *
 * 根据主题调用对应回调函数，若未找到主题回调则调用默认回调
 */
void MQTTClient::onMessage(const struct mosquitto_message *msg) {
  std::string topic = msg->topic ? msg->topic : "";
  std::string payload((char *)msg->payload, msg->payloadlen);

  auto it = m_topic_callbacks.find(topic);
  if (it != m_topic_callbacks.end()) {
    it->second(topic, payload);
  } else if (m_default_cb) {
    m_default_cb(topic, payload);
  }
}

} // namespace mqtt
